package com.semidata.trp.frequency.redis;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

import org.aredis.cache.AsyncHandler;
import org.aredis.cache.AsyncRedisConnection;
import org.aredis.cache.RedisCommandInfo;
import org.aredis.cache.RedisCommandInfo.CommandStatus;
import org.aredis.net.ConnectionStatus;

import com.semidata.rtd.core.datasource.DataSourceCommon;
import com.semidata.rtd.core.datasource.DataSourceCompleteHandler;
import com.semidata.rtd.core.datasource.DataSourceErrorHandler;
import com.semidata.rtd.core.datasource.redis.RedisDataSource;
/*
class RedisCommon implements DataSourceCommon {
	private enum Status {
		INPROCESS, CANCELLED, DONE,
	}

	protected RedisDataSource datasource;
	protected Map<AsyncRedisConnection, List<RedisCommandInfo>> clientMap;
	protected Map<String, Map<String, Object>> resultMap;
	protected List<Future<RedisCommandInfo[]>> futureList;
	protected AtomicReference<Status> status;
	protected AtomicLong totalCommands;
	private int timeout = 1000;
	private DataSourceCompleteHandler completeHandler;
	private DataSourceErrorHandler errorHandler;

	public RedisCommon(RedisDataSource datasource) {
		this.datasource = datasource;
		this.totalCommands = new AtomicLong(0);
		this.status = new AtomicReference<Status>(Status.INPROCESS);
		this.clientMap = new HashMap<AsyncRedisConnection, List<RedisCommandInfo>>();
		this.resultMap = new ConcurrentHashMap<String, Map<String, Object>>();
		this.futureList = new ArrayList<Future<RedisCommandInfo[]>>();
	}

	public void setTimeout(int timeout) {
		this.timeout = timeout;
	}

	public void submit() {

		for (AsyncRedisConnection client : clientMap.keySet()) {
			List<RedisCommandInfo> list = clientMap.get(client);

			RedisCommandInfo[] array = new RedisCommandInfo[list.size()];
			list.toArray(array);

			client.getConnection().getConfig().setReadTimeoutMillis(timeout);
			client.getConnection().getConfig().setWriteTimeoutMillis(timeout);

			// final long start = System.currentTimeMillis();
			Future<RedisCommandInfo[]> future = client.submitCommands(array,

			new AsyncHandler<RedisCommandInfo[]>() {

				public void completed(RedisCommandInfo[] result, Throwable e) {

					if (e != null) {
						if (errorHandler != null)
							errorHandler.handle(e);
					}

					// long end =System.currentTimeMillis();
					// System.out.println("read data time : " + (end - start));
					RedisCommandInfo oneResult = result[0];
					AsyncRedisConnection conn = datasource.getClient(
					        oneResult.getParams()[0].toString(), "");

					for (RedisCommandInfo info : result) {
						Object[] params = info.getParams();
						String command = info.getCommand().toString();

						if (command.equals("EXPIRE") || command.equals("HSET"))
							continue;
						String database = (String) params[0];

						Object[] values = (Object[]) info.getResult();
						CommandStatus status = info.getRunStatus();
						if (status.ordinal() != CommandStatus.SUCCESS.ordinal()) {
							// System.out.println(status.name());
							continue;
						}
						Map<String, Object> map = resultMap.get(database);
						if (map == null) {
							map = new HashMap<String, Object>();
							resultMap.put(database, map);
						}
						String key = "";
						String value = "";
						for (int i = 1; i <= values.length; i++) {
							if (i % 2 == 0) {
								value = values[i - 1].toString();
								if (conn.getConnection().getStatus() != ConnectionStatus.OK)
									map.put(key, null);
								else
									map.put(key, value);
							} else {
								key = values[i - 1].toString();
							}
						}

					}

					long remaining = totalCommands.addAndGet(-result.length);
					if (remaining == 0) {
						if (status.compareAndSet(Status.INPROCESS, Status.DONE)) {
							if (completeHandler != null)
								completeHandler.handle();
						}
					}
				}
			}, true, false);
			futureList.add(future);
		}
	}

	public boolean cancel(boolean mayInterruptIfRunning) {

		if (!status.compareAndSet(Status.INPROCESS, Status.CANCELLED))
			return false;

		for (Future<RedisCommandInfo[]> future : futureList) {
			future.cancel(mayInterruptIfRunning);
		}

		if (errorHandler != null)
			errorHandler.handle(null);
		return true;
	}

	public boolean isCancelled() {
		return status.get() == Status.CANCELLED;
	}

	public boolean isDone() {
		return status.get() == Status.DONE;
	}

	public void setCompleteHandler(DataSourceCompleteHandler completeHandler) {
		this.completeHandler = completeHandler;
	}
	
	public void setErrorHandler(DataSourceErrorHandler errorHandler) {
		this.errorHandler = errorHandler;
	}

	public Object get() throws InterruptedException, ExecutionException {
		// TODO Auto-generated method stub
		throw new UnsupportedOperationException();
	}

	public Object get(long timeout, TimeUnit unit) throws InterruptedException,
	        ExecutionException, TimeoutException {
		// TODO Auto-generated method stub
		throw new UnsupportedOperationException();

	}

}
*/